package com.klm.easymq.core.adapter.impl;

import com.klm.easymq.core.adapter.MQAdapter;
import com.klm.easymq.core.adapter.config.MQConfig;
import com.klm.easymq.core.adapter.config.ConsumerConfig;
import com.klm.easymq.core.adapter.message.Message;
import com.klm.easymq.metrics.EasyMqMetrics;
import io.micrometer.core.instrument.Timer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RocketMQ适配器实现
 * 将RocketMQ功能适配到统一的MQ适配器接口
 */
@Slf4j
@Component
public class RocketMQAdapter implements MQAdapter {
    
    private static final String ADAPTER_TYPE = "rocketmq";
    
    @Autowired(required = false)
    private RocketMQTemplate rocketMqTemplate;
    
    @Autowired(required = false)
    private EasyMqMetrics metrics;
    
    private MQConfig config;
    private final Map<String, Object> consumers = new ConcurrentHashMap<>();
    private AdapterStatus status = AdapterStatus.UNINITIALIZED;
    
    @Override
    public String getType() {
        return ADAPTER_TYPE;
    }
    
    @Override
    public void init(MQConfig config) {
        this.config = config;
        this.status = AdapterStatus.INITIALIZED;
        log.info("RocketMQ适配器初始化完成: {}", config.getAddress());
        
        // 更新连接状态指标
        if (metrics != null) {
            metrics.setConnectionStatus(ADAPTER_TYPE, true);
        }
    }
    
    @Override
    public boolean sendSync(Message message) {
        Timer.Sample timer = null;
        if (metrics != null) {
            timer = metrics.startMessageSendTimer();
        }
        
        try {
            if (rocketMqTemplate == null) {
                log.error("RocketMQTemplate未初始化");
                if (metrics != null) {
                    metrics.incrementError(ADAPTER_TYPE, "template_not_initialized");
                }
                return false;
            }
            
            org.springframework.messaging.Message<String> msg = buildSpringMessage(message);
            SendResult sendResult = rocketMqTemplate.syncSend(message.getTopic(), msg);
            
            log.info("RocketMQ同步消息发送成功: topic={}, messageId={}, tag={}, key={}", 
                    message.getTopic(), sendResult.getMsgId(), message.getTag(), message.getKey());
            
            // 记录成功指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), true);
            }
            
            return true;
            
        } catch (Exception e) {
            log.error("RocketMQ同步消息发送失败: topic={}, message={}", message.getTopic(), message.getBody(), e);
            
            // 记录失败指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), false);
                metrics.incrementError(ADAPTER_TYPE, "send_failed");
            }
            
            return false;
        } finally {
            // 停止计时器
            if (timer != null && metrics != null) {
                boolean success = rocketMqTemplate != null;
                metrics.stopMessageSendTimer(timer, ADAPTER_TYPE, message.getTopic(), success);
            }
        }
    }
    
    @Override
    public CompletableFuture<Boolean> sendAsync(Message message) {
        Timer.Sample timer = null;
        if (metrics != null) {
            timer = metrics.startMessageSendTimer();
        }
        
        CompletableFuture<Boolean> future = new CompletableFuture<>();
        
        try {
            if (rocketMqTemplate == null) {
                log.error("RocketMQTemplate未初始化");
                if (metrics != null) {
                    metrics.incrementError(ADAPTER_TYPE, "template_not_initialized");
                }
                future.complete(false);
                return future;
            }
            
            org.springframework.messaging.Message<String> msg = buildSpringMessage(message);
            rocketMqTemplate.asyncSend(message.getTopic(), msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("RocketMQ异步消息发送成功: topic={}, messageId={}, tag={}, key={}", 
                            message.getTopic(), sendResult.getMsgId(), message.getTag(), message.getKey());
                    
                    // 记录成功指标
                    if (metrics != null) {
                        metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), true);
                    }
                    
                    future.complete(true);
                }
                
                @Override
                public void onException(Throwable throwable) {
                    log.error("RocketMQ异步消息发送失败: topic={}, message={}", 
                            message.getTopic(), message.getBody(), throwable);
                    
                    // 记录失败指标
                    if (metrics != null) {
                        metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), false);
                        metrics.incrementError(ADAPTER_TYPE, "async_send_failed");
                    }
                    
                    future.complete(false);
                }
            });
            
        } catch (Exception e) {
            log.error("RocketMQ异步消息发送异常: topic={}, message={}", message.getTopic(), message.getBody(), e);
            
            // 记录失败指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), false);
                metrics.incrementError(ADAPTER_TYPE, "async_send_exception");
            }
            
            future.complete(false);
        } finally {
            // 停止计时器
            if (timer != null && metrics != null) {
                boolean success = rocketMqTemplate != null;
                metrics.stopMessageSendTimer(timer, ADAPTER_TYPE, message.getTopic(), success);
            }
        }
        
        return future;
    }
    
    @Override
    public void sendOneway(Message message) {
        try {
            if (rocketMqTemplate == null) {
                log.error("RocketMQTemplate未初始化");
                if (metrics != null) {
                    metrics.incrementError(ADAPTER_TYPE, "template_not_initialized");
                }
                return;
            }
            
            org.springframework.messaging.Message<String> msg = buildSpringMessage(message);
            rocketMqTemplate.sendOneWay(message.getTopic(), msg);
            
            log.info("RocketMQ单向消息发送成功: topic={}, tag={}, key={}", 
                    message.getTopic(), message.getTag(), message.getKey());
            
            // 记录成功指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), true);
            }
            
        } catch (Exception e) {
            log.error("RocketMQ单向消息发送失败: topic={}, message={}", message.getTopic(), message.getBody(), e);
            
            // 记录失败指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), false);
                metrics.incrementError(ADAPTER_TYPE, "oneway_send_failed");
            }
        }
    }
    
    @Override
    public boolean sendDelay(Message message, int delayLevel) {
        Timer.Sample timer = null;
        if (metrics != null) {
            timer = metrics.startMessageSendTimer();
        }
        
        try {
            if (rocketMqTemplate == null) {
                log.error("RocketMQTemplate未初始化");
                if (metrics != null) {
                    metrics.incrementError(ADAPTER_TYPE, "template_not_initialized");
                }
                return false;
            }
            
            org.springframework.messaging.Message<String> msg = buildSpringMessage(message);
            SendResult sendResult = rocketMqTemplate.syncSend(message.getTopic(), msg, 
                    config.getRequestTimeout(), delayLevel);
            
            log.info("RocketMQ延迟消息发送成功: topic={}, messageId={}, delayLevel={}", 
                    message.getTopic(), sendResult.getMsgId(), delayLevel);
            
            // 记录成功指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), true);
            }
            
            return true;
            
        } catch (Exception e) {
            log.error("RocketMQ延迟消息发送失败: topic={}, message={}, delayLevel={}", 
                    message.getTopic(), message.getBody(), delayLevel, e);
            
            // 记录失败指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), false);
                metrics.incrementError(ADAPTER_TYPE, "delay_send_failed");
            }
            
            return false;
        } finally {
            // 停止计时器
            if (timer != null && metrics != null) {
                boolean success = rocketMqTemplate != null;
                metrics.stopMessageSendTimer(timer, ADAPTER_TYPE, message.getTopic(), success);
            }
        }
    }
    
    @Override
    public boolean sendOrdered(Message message, String hashKey) {
        Timer.Sample timer = null;
        if (metrics != null) {
            timer = metrics.startMessageSendTimer();
        }
        
        try {
            if (rocketMqTemplate == null) {
                log.error("RocketMQTemplate未初始化");
                if (metrics != null) {
                    metrics.incrementError(ADAPTER_TYPE, "template_not_initialized");
                }
                return false;
            }
            
            String messageBody = message.getBody() != null ? message.getBody().toString() : "";
            SendResult sendResult = rocketMqTemplate.syncSendOrderly(message.getTopic(), messageBody, hashKey);
            
            log.info("RocketMQ顺序消息发送成功: topic={}, messageId={}, hashKey={}", 
                    message.getTopic(), sendResult.getMsgId(), hashKey);
            
            // 记录成功指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), true);
            }
            
            return true;
            
        } catch (Exception e) {
            log.error("RocketMQ顺序消息发送失败: topic={}, message={}, hashKey={}", 
                    message.getTopic(), message.getBody(), hashKey, e);
            
            // 记录失败指标
            if (metrics != null) {
                metrics.incrementMessageSent(ADAPTER_TYPE, message.getTopic(), false);
                metrics.incrementError(ADAPTER_TYPE, "ordered_send_failed");
            }
            
            return false;
        } finally {
            // 停止计时器
            if (timer != null && metrics != null) {
                boolean success = rocketMqTemplate != null;
                metrics.stopMessageSendTimer(timer, ADAPTER_TYPE, message.getTopic(), success);
            }
        }
    }
    
    @Override
    public List<Boolean> sendBatch(List<Message> messages) {
        // RocketMQ的批量发送需要特殊处理，这里简化为循环发送
        List<Boolean> results = messages.stream()
                .map(this::sendSync)
                .collect(java.util.stream.Collectors.toList());
        
        // 记录批量发送指标
        if (metrics != null) {
            long successCount = results.stream().mapToLong(result -> result ? 1 : 0).sum();
            boolean allSuccess = successCount == messages.size();
            metrics.recordBatchSend(ADAPTER_TYPE, "batch", messages.size(), allSuccess);
        }
        
        return results;
    }
    
    @Override
    public void startConsumer(ConsumerConfig consumerConfig) {
        // RocketMQ的消费者启动需要特殊的注解配置
        // 这里只是记录配置，实际的消费者启动由Spring容器管理
        String consumerId = consumerConfig.getConsumerId();
        consumers.put(consumerId, consumerConfig);
        log.info("RocketMQ消费者配置已记录: consumerId={}, topics={}", 
                consumerId, consumerConfig.getTopics());
    }
    
    @Override
    public void stopConsumer(String consumerId) {
        consumers.remove(consumerId);
        log.info("RocketMQ消费者已停止: consumerId={}", consumerId);
    }
    
    @Override
    public boolean isHealthy() {
        // 简单的健康检查，检查RocketMQTemplate是否存在
        boolean healthy = rocketMqTemplate != null && status != AdapterStatus.ERROR;
        
        // 更新连接状态指标
        if (metrics != null) {
            metrics.setConnectionStatus(ADAPTER_TYPE, healthy);
        }
        
        return healthy;
    }
    
    @Override
    public void shutdown() {
        this.status = AdapterStatus.SHUTDOWN;
        consumers.clear();
        
        // 更新连接状态指标
        if (metrics != null) {
            metrics.setConnectionStatus(ADAPTER_TYPE, false);
        }
        
        log.info("RocketMQ适配器已关闭");
    }
    
    @Override
    public AdapterStatus getStatus() {
        return status;
    }
    
    /**
     * 构建Spring Message对象
     * @param message 统一消息对象
     * @return Spring Message对象
     */
    private org.springframework.messaging.Message<String> buildSpringMessage(Message message) {
        String messageBody = message.getBody() != null ? message.getBody().toString() : "";
        
        MessageBuilder<String> builder = MessageBuilder.withPayload(messageBody);
        
        if (message.getTag() != null && !message.getTag().isEmpty()) {
            builder.setHeader("tag", message.getTag());
        }
        
        if (message.getKey() != null && !message.getKey().isEmpty()) {
            builder.setHeader("key", message.getKey());
        }
        
        // 添加自定义属性
        if (message.getProperties() != null) {
            for (Map.Entry<String, String> entry : message.getProperties().entrySet()) {
                builder.setHeader(entry.getKey(), entry.getValue());
            }
        }
        
        return builder.build();
    }
} 